import java.io.IOException;
import java.util.Set;

import com.google.gson.Gson;
import com.google.gson.JsonSyntaxException;
import com.jiuan.transfer.mqtt.MqttUtil;
import com.jiuan.transfer.mqtt.domain.Config;
import com.jiuan.transfer.mqtt.domain.MsgBean;
import com.jiuan.transfer.rabbitmq.ChannelFactory;
import com.jiuan.transfer.redis.SingleRedisUtil;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

/**
 * Created by Administrator on 2017/1/4.
 */
public class StarIt2 {
    public static void main(String[] args) {


        // 初始化配置文件
        Config.initConfig(null);

        // 创建rabbitmq队列
        declareQueue(Config.getRabbit_queue());

        // 创建rabbitmq消费者线程
        try {
            initThreadPool(Config.getRabbit_queue(),
                    Config.getRabbit_connection_number());
        } catch (Exception e) {
            e.printStackTrace();
            System.exit(1);

        }

        //启动完毕
        System.out.println("=============================================");

    }


    private static void declareQueue(String queue) {
        com.rabbitmq.client.Channel channel = null;
        try {
            channel = ChannelFactory.getChannel();
            channel.queueDeclare(queue, true, false, false, null);
            channel.close();
        } catch (Exception e) {

            e.printStackTrace();
            System.out.println("RabbitMQ queue declare failed, system exit!");
            System.exit(0);
        }

    }

    private static void initThreadPool(String queue, int size) throws Exception {
        for (int i = 0; i < size; i++) {
            try {
                final com.rabbitmq.client.Channel channel = ChannelFactory
                        .getChannel();
                final Gson gson = new Gson();
                boolean autoAck = false;
                channel.basicConsume(queue, autoAck, new DefaultConsumer(
                        channel) {
                    @Override
                    public void handleDelivery(String consumerTag,
                                               Envelope envelope, AMQP.BasicProperties properties,
                                               byte[] body) throws IOException {

                        long deliveryTag = envelope.getDeliveryTag();

                        // 收到message
                        String message = new String(body, "UTF-8");
                        MsgBean msgBean = null;

                        try {
                            msgBean = gson.fromJson(message, MsgBean.class);

                        } catch (JsonSyntaxException e) {
                            e.printStackTrace();
                            System.err.println("**********消息格式錯誤************");
                            return;
                        }
                        System.out.println();
                        System.out.println();
                        System.out.println(message);
                        System.out.println("RabbitMQ   [x] Received:     " + msgBean);


                        // 獲取主題和消息内容
                        String topic = msgBean.getTopic();
                        byte[] publishmessage = (gson.toJson(msgBean)).getBytes("UTF-8");

                        // 查询推送ip
                        Set<String> ips = SingleRedisUtil.queryIps("TOPIC_"
                                + topic);

                        // 发布主题MQQT
                        try {
                            msgBean.setTopic(null);//无关字段
                            publish(topic, ips, publishmessage);
                        } catch (Exception e) {
                            System.err.println("\tMQTT publish Failed :"+e.getMessage());
                            e.printStackTrace();

                            //channel.basicNack(deliveryTag, false, true);//未应答消息重新进入队列
                        }finally{
                            channel.basicAck(deliveryTag, false);
                        }


                    }

                });
                System.out.println(Thread.currentThread().getName()
                        + ">>   RabbitMq 订阅OK!");
            } catch (IOException e) {
                System.err.println("RabbitMq 订阅失败");
                e.printStackTrace();
            }catch (Exception e) {
                System.err.println("Exception:::");
                e.printStackTrace();
            }
        }
    }

    // 发布mqtt消息
    private static void publish(String topic, Set<String> queryIps,
                                byte[] message) throws Exception  {
        if (null == queryIps || queryIps.size() <= 0)
            return;
        if (null == message || message.length <= 0)
            return;
        if (null == topic || "".equals(topic))
            return;

        for (String ip : queryIps) {
            System.out.println("\tMQTT publish... :  Topic: "+topic+"    IP: "+ip);
            MqttUtil.publish(ip, topic, message);
        }
    }
}
